package com.atguigu.app

import com.alibaba.fastjson.JSON
import com.atguigu.beans.{CouponAlertInfo, EventLog}
import com.atguigu.gmall.constances.CommonConstants
import com.atguigu.util.MyEsUtil
import com.atguigu.util.MykafkaUtil.MyKafkaUtil
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import java.time.format.DateTimeFormatter
import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
import java.util
import scala.util.control.Breaks.{break, breakable}

object AlertApp extends BaseApp {
  override var appName: String = "AlertApp"
  override var internal: Int = 5

  def main(args: Array[String]): Unit = {

    sc=new StreamingContext("local[4]",appName,Seconds(internal))

    runApp{
      //消费kafka,获取KAFKA-EVENTLOG主题的数据
      val ds = MyKafkaUtil.getKafkaStream(CommonConstants.KAFKA_EVENT_LOG, sc)
      // 1.json数据封装成eventlog
      val ds1 = ds.map(log => {
        val log1 = JSON.parseObject(log.value(), classOf[EventLog])
        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
        val time: LocalDateTime = LocalDateTime.ofInstant(Instant.ofEpochSecond(log1.ts), ZoneId.of("Asia/Shanghai"))
        log1.logHour=time.getHour+""
        log1.logDate=time.format(formatter)
        log1
      })

      // 2.开窗
      val ds2 = ds1.window(Minutes(5))
      // 3.map转换每条数据为设备,用户为k的格式,便于后续分组
      val ds3 = ds2.map(log => {
        ((log.mid, log.uid), log)
      })
      // 4. 每个设备和每个用户一组,值为这个用户的一堆记录
      val ds4 = ds3.groupByKey()
      // 5. 开始对每组进行过滤,过滤出领取了优惠券,没有浏览商品的用户,使用打标记的方式进行过滤,为什么不用filter?因为filter是过滤
      //logs中不符合条件的log,而这里是只要有不符合条件的log就把整个用户给过滤掉
      val ds5 = ds4.map {
        case ((mid, uid), logs) => {
          var isNeedAlert: Boolean = false
          breakable {
            logs.foreach(log => {

              if ("clickItem" == log.evid) {
                break()
              } else if ("coupon" == log.evid) {
                isNeedAlert = true //2.如果没有浏览商品,只要领取了优惠券,就预警
              }
            })
          } //1.只要有一条日志是浏览商品就不用判断了,直接跳出遍历
          //3.遍历完这个用户了,判断是否需要预警
          if (isNeedAlert) {
            (mid, logs)
          } else {
            (null, null)
          }
        }
      }
      //6.过滤掉null的数据
      val ds6 = ds5.filter(_._1 != null)
      //7. 设备分组,过滤>=3条的数据
      val ds7 = ds6.groupByKey()
      val ds8 = ds7.filter(_._2.size >= 3)
      //user1(log1,log2,log3),user2(log1,log2,log3)
      val ds9 = ds8.mapValues(_.flatten)

      //8. 准备将过滤出的日志发往es: 先生成特定格式的预警日志
      val ds10 = ds9.map {
        case (mid, logs) => {
          var uids: util.HashSet[String] = new util.HashSet[String]
          var itemIds: util.HashSet[String] = new util.HashSet[String]
          var events: util.List[String] = new util.ArrayList[String]
          logs.foreach(log => {
            uids.add(log.uid)
            itemIds.add(log.itemid)
            events.add(log.evid)
          })
          CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis())
        }
      }
      //9.再转换为工具类要求的格式,并且根据需求同一设备每分钟最多产生一条数据,将设备+分钟拼接为id

      //List(String,Any)  String : id  Any : 一行数据

        //因为需求是同一设备每分钟最多产生一条数据,所以id应该为设备+分钟

      val ds11 = ds10.map(couponAlertInfo => {
        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm")
        val time = LocalDateTime.ofInstant(Instant.ofEpochMilli(couponAlertInfo.ts), ZoneId.of("Asia/Shanghai"))
        val str = time.format(formatter)
        (couponAlertInfo.mid + "-" + str, couponAlertInfo) //注意es中属性类型为keyword,样例类类型属性类型为集合,但不用担心,会自动转换
      })
      ds11.count().print(1000)
      //10 写入es
      ds11.foreachRDD(rdd=>{
        rdd.foreachPartition(partition=>{
          MyEsUtil.insertBulk("gmall_coupon_alert"+LocalDate.now(),partition.toList)
        })
      })

    }
  }
}
